package com.atbeijing.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object SparkStreaming02_Source_Queue {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")

        // 构建StreamingContext对象时，第二个参数表示数据 采集周期，以毫秒单位，一般以秒为单位使用
        val ssc = new StreamingContext(sparkConf, Seconds(3))

        val rddQueue = new mutable.Queue[RDD[Int]]()
        val queueDS = ssc.queueStream(rddQueue, oneAtATime = false)

        // TODO No output operations registered, so nothing to execute
        //println(queueDS)打印队列中的1000个数据
        queueDS.print(1000)

        // 启动采集器
        ssc.start()

        //8.循环创建并向RDD队列中放入RDD,每个rdd为1-300,每10个一个分区
        for (i <- 1 to 5) {
            rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
            Thread.sleep(2000)
        }

        ssc.awaitTermination()
    }
}
